package cn.lsh.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;

public class WordCount {

	public static void main(String[] args) {
		/**
		 * conf
		 *   1、可以设置spark的运行模式
		 *   2、可以设置spark在webui中显示的application的名称
		 *   3、可以设置集群spark application运行的资源（内存+core）
		 *
		 * spark运行模式
		 * 	 1、local，本地运行
		 * 	 2、standalone，spark自带资源调度框架，spark://node0:7077
		 * 	 3、yarn，hadoop的资源调度框架，仅支持spark-submit提交方式
		 * 	 4、mesos，资源调度框架
		 */
		SparkConf conf = new SparkConf();
		conf.setMaster("local").setAppName("wordCount");

		/** SparkContext是通过集群的唯一通道 */
		JavaSparkContext sc = new JavaSparkContext(conf);
		//读取文件
		JavaRDD<String> stringJavaRDD = sc.textFile("hdfs://mycluster/test/input/firend/friden_list.txt");
		//flatMap合并流为一个RDD
		JavaRDD<String> wordJavaRDD = stringJavaRDD.flatMap((FlatMapFunction<String, String>) line -> Arrays.asList(line.split(" ")).iterator());
		//mapToPair转换成二元组RDD
		JavaPairRDD<String, Integer> tupleJavaPairRDD = wordJavaRDD.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<>(word, 1));
		//Function2<T1, T2, R>：T1为value1，T2为value2，R为返回；根据二元组key分组统计
		JavaPairRDD<String, Integer> resultJavaPairRDD = tupleJavaPairRDD.reduceByKey((v1, v2) -> v1 + v2);
		// long count = tupleJavaPairRDD.reduceByKey(Integer::sum).count();
		//遍历输出，入参为Tuple2。Tuple2::swap 元组key和value变换位置，sortByKey根据元组的key排序
		resultJavaPairRDD.mapToPair(Tuple2::swap).sortByKey(false).
				foreach(t-> System.out.println("word：" + t._2() + "， count：" + t._1()));
		sc.stop();
	}

}
